package com.klikmee.sms.core.commons.queues;

import java.text.MessageFormat;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import com.klikmee.sms.core.commons.events.AppEvent;
import com.klikmee.sms.core.commons.events.Dispatcher;
import com.klikmee.sms.core.commons.exceptions.ErrorInfo;
import com.klikmee.sms.core.commons.exceptions.ExceptionHandler;
import com.klikmee.sms.core.commons.exceptions.ErrorInfo.ErrorType;
import com.klikmee.sms.core.commons.exceptions.ErrorInfo.Severity;

/**
 * Simple queue processor implementation.
 * 
 * @author makcro
 * 
 */
public class SimpleQueueManagerImpl<T> implements QueueManager<T> {

	// logger instance for tihs processor
	private Logger						logger			= Logger.getLogger( SimpleQueueManagerImpl.class );

	// queue containing items for processing
	private BlockingQueue<QueueItem<T>>	queue;

	/**
	 * Service used to execute task of queue processing
	 */
	private ScheduledExecutorService	executorService;

	/**
	 * Item processor for queue items
	 */
	private QueueItemProcessor<T>		itemProcessor;

	/**
	 * Instance of exception handler used to log error messages
	 */
	private ExceptionHandler			exceptionHandler;

	/**
	 * Number of mills to wait before re-putting item to the queue
	 */
	private Long						requeuePause	= null;

	/**
	 * Event dispatcher instance
	 */
	protected Dispatcher dispatcher;
	
	public SimpleQueueManagerImpl() {
		// detault requePause
		requeuePause = new Long( 0 );
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.klikmee.sms.core.commons.queues.QueueManager#enqueue(com.klikmee.
	 * sms.core.commons.queues.QueueItem)
	 */
	@Override
	public void enqueue(QueueItem<T> queueItem) throws EnqueueItemException {
		try {
			this.queue.put( queueItem );
			this.logger.info( MessageFormat.format( "Enqueued new item, desc [{0}]", queueItem ) );
		}
		catch (InterruptedException e) {
			logger.debug( "Error while putting item into queue, throwing ex" );

			EnqueueItemException enqueueItemException = new EnqueueItemException( "Could not enque item", e );
			ErrorInfo errorInfo = enqueueItemException.addInfo();
			errorInfo.setCause( e );
			errorInfo.setContextId( "QueueManager/EnquqeItem" );
			errorInfo.setErrorId( "EnqueueItem" );
			errorInfo.setErrorType( ErrorType.INTERNAL );
			errorInfo.setSeverity( Severity.FATAL );
			errorInfo.getParameters().put( "ITEM", queueItem );

			throw enqueueItemException;
		}

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.klikmee.sms.core.commons.queues.QueueManager#start()
	 */
	@Override
	public void start() {

		Runnable callable = new Runnable() {

			@Override
			public void run() {
				// run indefinitely
				while (true) {
					// reference to the item to process
					QueueItem<T> item = null;
					try {
						// take item from the queue, and delegate it to
						// processor
						item = SimpleQueueManagerImpl.this.queue.take();
						SimpleQueueManagerImpl.this.logger.debug( MessageFormat.format( "Item [{0}] taken from queue", item.toString() ) );
						boolean itemProcessed = SimpleQueueManagerImpl.this.itemProcessor.process( item );

						// if item was successfully processed, log it and take
						// another
						// else, consider reputing it into queue
						if (itemProcessed) {
							SimpleQueueManagerImpl.this.logger.debug( MessageFormat.format( "Successfully processed item [{0}]", item.toString() ) );
						}
						else {
							SimpleQueueManagerImpl.this.logger.debug( MessageFormat.format( "Processing of Item [{0}] failed, considering requeue...", item.toString() ) );
							boolean reQueueItem = considerItemReque( item );
							if (reQueueItem) {
								SimpleQueueManagerImpl.this.requeue( item );
							}
							else {
								discardItem( item );
							}
						}
					}
					catch (InterruptedException e) {
						// this usually happens when queue manager is stopping
						SimpleQueueManagerImpl.this.logger.fatal( "Interrupted while waiting for queue element", e );
					}
					catch (Exception e) {
						// if exception occurs during processing, log it and
						SimpleQueueManagerImpl.this.logger.fatal( MessageFormat.format( "Error while processing item [{0}]", item != null ? item.toString() : "Unknown" ), e );
					}
				}

			}

		};

		executorService.schedule( callable, 0, TimeUnit.MILLISECONDS );
		logger.info( "Queue manager started..." );
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.klikmee.sms.core.commons.queues.QueueManager#stop()
	 */
	@Override
	public void stop() {
		this.executorService.shutdown();
		logger.info( "Queue manager stoped..." );
	}

	/**
	 * Decides if item should be re-queued or not
	 * 
	 * @param item
	 *            item to consider
	 * @return true if item should be re-queued, false otherwise
	 */
	protected boolean considerItemReque(QueueItem<T> item) {
		boolean requeue = false;
		// decide based on items validity
		if (new Date().before( item.getValidTo() )) {
			requeue = true;
		}
		return requeue;
	}

	@Override
	public void requeue(final QueueItem<T> item) throws EnqueueItemException {
		Callable<Void> callable = new Callable<Void>() {

			@Override
			public Void call() throws Exception {
				SimpleQueueManagerImpl.this.queue.put( item );
				SimpleQueueManagerImpl.this.logger.info( MessageFormat.format( "Item [{0}] re-put into queue", item ) );
				return null;
			}
		};

		ScheduledFuture<Void> future = this.executorService.schedule( callable, this.requeuePause, TimeUnit.SECONDS );

	}

	protected void discardItem(QueueItem<T> item) {
		logger.warn( MessageFormat.format( "Discarding item {0}", item ) );
		// fire item discarded event with item specified event type
		AppEvent appEvent = new AppEvent( item.getDiscardedEventType(), item );
		this.dispatcher.dispatch( appEvent );
	}

	@Override
	public void setItemProcessor(QueueItemProcessor<T> itemProcessor) {
		this.itemProcessor = itemProcessor;
	}

	public void setRequeuePause(Long requeuePause) {
		this.requeuePause = requeuePause;
	}

	public void setExecutorService(ScheduledExecutorService executorService) {
		this.executorService = executorService;
	}

	public void setExceptionHandler(ExceptionHandler exceptionHandler) {
		this.exceptionHandler = exceptionHandler;
	}

	public void setQueue(BlockingQueue<QueueItem<T>> queue) {
		this.queue = queue;
	}
	
	public void setDispatcher(Dispatcher dispatcher) {
		this.dispatcher = dispatcher;
	}

}
